package eiisan.parallelstream;

import com.alibaba.ttl.TransmittableThreadLocal;
import com.alibaba.ttl.TtlRunnable;
import com.alibaba.ttl.threadpool.TtlExecutors;
import com.alibaba.ttl.threadpool.TtlForkJoinPoolHelper;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ForkJoinPool;

public class ForkJoinPoolTest {
    //结论：java stream流并发执行线程分配
    //自定义线程池，直接将所有数据丢到自定义线程池中执行。
    //默认线程池，首次执行任务时会复用主线程main，(所有数据数量-1)个任务丢到线程池中执行。之后会全部丢到线程池中执行
    //参考https://blog.csdn.net/u010597819/article/details/111070454

    //parallelStream的问题
    //1.多个parallelStream之间默认使用的是同一个线程池，所以IO操作尽量不要放进parallelStream中，否则会阻塞其他parallelStream
    //2.系统中如果使用了会话管理器之类的东西,就是利用ThreadLocal 或 InheritableThreadLocal来制造了一个线程变量,存放每次请求的会话线程的线程变量，
    // 如果在parallelStream去取ThreadLocal的内容，可能为空的异常，因为parallelStream中的任务线程可能不是线程变量的子线程，导致意想不到的结果

    //parallelStream的问题



    //正确使用1
    public static void main1(String[] args) throws Exception {
        System.out.println("Main thread start");

        ExecutorService forkJoinExecutorService = new ForkJoinPool(4);
        forkJoinExecutorService = TtlExecutors.getTtlExecutorService(forkJoinExecutorService);
        TransmittableThreadLocal<String> stringTransmittableThreadLocal = new TransmittableThreadLocal<String>();
        // 每个租户的数据，暂定写死咯
        int size = 10;
        List<String> data = new ArrayList<>(size);
        for (int i = 0; i < size; i++) {
            data.add(UUID.randomUUID().toString());
        }
        // 3个租户
        for (String s : new String[]{"ZHANGSAN", "LISI", "WANGWU"}) {
            System.out.println(Thread.currentThread().getName() + " " + s + "---" + stringTransmittableThreadLocal.get());
            // 1. 按照租户分片
            // 2. 设置租户ID，对每个并发线程共享
            stringTransmittableThreadLocal.set(s);

            try {
                Runnable task = new Runnable() {
                    @Override
                    public void run() {
                        System.out.println(Thread.currentThread().getName() + " start");

                        // 3. 并发执行业务逻辑：根据租户ID获取需要预热的数据，进行缓存预热
                        data.stream()
                                .forEach(t -> {
                                    System.out.println(Thread.currentThread().getName() + "---" + t + "---" + stringTransmittableThreadLocal.get());
                                    try {
                                        Thread.sleep(100);
                                    } catch (InterruptedException e) {
                                        e.printStackTrace();
                                    }
                                });

                        System.out.println(Thread.currentThread().getName() + " end and clear threadLocal");
                    }
                };

                //每次提交时都需要通过修饰操作（即TtlRunnable.get(task)）以抓取这次提交时的TransmittableThreadLocal上下文的值
                // TtlRunnable ttlRunnable = TtlRunnable.get(task);
                forkJoinExecutorService.submit(task).get();
            } finally {
                // 4. 清理租户ID
                stringTransmittableThreadLocal.remove();
            }
            Thread.sleep(1000);
            System.out.println(s + " end !!!");
            System.out.println(Thread.currentThread().getName() + "--- end ---" + stringTransmittableThreadLocal.get());
        }
    }

    //这里就是错误例子！！！！！！！，parallelStream里面threadLocal.get数据是错乱的
    public static void mainerror(String[] args) throws Exception {
        System.out.println("Main thread start");
        ForkJoinPool forkJoinPool = new ForkJoinPool(4);
//这里就是错误例子！
        InheritableThreadLocal<String> threadLocal = new InheritableThreadLocal<>();
        // 每个租户的数据，暂定写死咯
        int size = 10;
        List<String> data = new ArrayList<>(size);
        for (int i = 0; i < size; i++) {
            data.add(UUID.randomUUID().toString());
        }
//这里就是错误例子！
        // 3个租户
        for (String s : new String[]{"ZHANGSAN", "LISI", "WANGWU"}) {
            System.out.println(Thread.currentThread().getName() + " " + s + "---" + threadLocal.get());
            // 1. 按照租户分片
            forkJoinPool.submit(() -> {
                System.out.println(Thread.currentThread().getName() + " start");
                try {
                    // 2. 设置租户ID，对每个并发线程共享
                    threadLocal.set(s);
                    // 3. 并发执行业务逻辑：根据租户ID获取需要预热的数据，进行缓存预热
                    data.stream().parallel().forEach(t -> {
                        System.out.println(Thread.currentThread().getName() + "---" + t + "---" + threadLocal.get());
                        try {
                            Thread.sleep(100);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    });
                } finally {
                    // 4. 清理租户ID
                    System.out.println(Thread.currentThread().getName() + " end and clear threadLocal");
                    threadLocal.remove();
                }
            }).get();
            Thread.sleep(1000);
            System.out.println(s + " end !!!");
            System.out.println(Thread.currentThread().getName() + "--- end ---" + threadLocal.get());
        }
    }
}
